// Copyright (c) 2021 Presto Labs Pte. Ltd.
// Author: gekim

#include "coin2/exchange/bitbank/feed_v1/processor.h"

#include <string>
#include <utility>

#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/bitbank/feed_v1/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::bitbank::feed_v1::impl {

void BitbankFeedProcessor::OnTopicRecord(const base::executor::TopicRecordPacket& packet) {
  upd_mgr_.onBeginPacket(packet);

  // Possible update messages:
  // 0{"sid":"8yM4H28PpXAFmLVCBskP","upgrades":[],"pingInterval":25000,"pingTimeout":5000}
  // 40
  // 42["message",{"room_name":"depth_diff_eth_jpy","message":{"data":{"a":[["215068","0.0218"],["214596","0"],["214613","0"],["214850","0"],["215139","1.194"],["214660","0.7095"]],"b":[["214437","0"],["214438","2"],["213828","1.194"],["214419","0"]],"t":1613803073181,"s":"1892756412"}}}]
  // 42["message",{"room_name":"depth_whole_mona_jpy","message":{"data":{"asks":[["287.896","1136.3407"],["287.900","2742.1469"],["287.902","3000.0000"],["287.970","2565.2890"],["288.213","14.7006"],["288.888","300.0000"],["288.900","1000.0000"],["289.200","3007.1253"],["289.203","69.6762"],["289.400","6745.4688"]...
  // 42["message",{"room_name":"transactions_ltc_jpy","message":{"data":{"transactions":[{"transaction_id":1044633718,"side":"sell","price":"25785.0","amount":"0.2327","executed_at":1613803070870},{"transaction_id":1044633719,"side":"sell","price":"25783.0","amount":"0.1553","executed_at":1613803070870}]}}}
  const auto& data = packet.payload;
  // Heart
  if (data == "3" || data == "40" || data == "41" || data.substr(0, 2) == "0{") {
    return;
  }

  rapidjson::Document document;
  auto pos = data.find_first_of('[');
  auto pos_json = data.find_first_of('{');
  if (pos_json == 0) {
    // maybe rest api response
    pos = 0;
  } else if (pos == std::string::npos || pos >= 3) {
    LOG(INFO) << "Not target JSON message: " << packet.payload;
    return;
  }

  document.Parse(data.c_str() + pos);
  if (document.HasParseError()) {
    THROW() << "JSON parse error: " << GetParseError_En(document.GetParseError()) << packet.payload;
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  if (document.IsArray()) {
    auto items = document.GetArray();
    CHECK_EQ_THROW(items.Size(), 2)
        << "[Bitbank Feed Processor] unexpected list length: " << packet.payload;
    if (items[1].HasMember("room_name") && items[1].HasMember("message")) {
      const auto& channel = BitbankChannel::Parse(items[1]["room_name"].GetString());

      if (!CheckChannelFilter(channel.channel)) {
        return;
      }

      const auto& native_symbol = channel.native_symbol;

      if (channel.channel == "transactions") {
        if (!upd_mgr_.is_symbol_registered(native_symbol)) {
          return;
        }

        Trade* agg_trade = upd_mgr_.GetTradeMutable();
        agg_trade->Clear();
        Trade trade{};

        const auto& trades = items[1]["message"]["data"]["transactions"].GetArray();
        agg_trade->timestamp =
            static_cast<int64_t>(trades[0]["executed_at"].GetInt64() * 1'000'000LL);
        for (auto& trade_json : trades) {
          BitbankFeedUpdateParser::ParseTrade(trade_json, &trade);
          if (TryAggregateTrades(agg_trade, trade)) {
            continue;
          }
          if (agg_trade->fill_qty > 0) {
            upd_mgr_.PublishTrade(native_symbol, true);
          }
          *agg_trade = trade;
        }
        if (agg_trade->fill_qty > 0) {
          upd_mgr_.PublishTrade(native_symbol, false);
        }
      } else if (channel.channel == "depth_diff") {
        auto* book = upd_mgr_.GetBookMutable(native_symbol);
        if (book != nullptr) {
          if (book->BidSize() > 0 && book->AskSize() > 0) {
            auto timestamp =
                static_cast<int64_t>(items[1]["message"]["data"]["t"].GetInt64() * 1'000'000LL);
            bool book_valid =
                BitbankFeedUpdateParser::ParseUpdate(book, items[1]["message"]["data"], timestamp);
            if (book_valid) {
              upd_mgr_.PublishBook();
            }
          }
        }
      } else if (channel.channel == "depth_whole") {
        auto* book = upd_mgr_.GetBookMutable(native_symbol);
        if (book != nullptr) {
          if (book->BidSize() == 0 || book->AskSize() == 0) {
            auto timestamp = static_cast<int64_t>(
                items[1]["message"]["data"]["timestamp"].GetInt64() * 1'000'000LL);
            bool book_valid = BitbankFeedUpdateParser::ParseSnapshot(
                book,
                items[1]["message"]["data"],
                timestamp);
            if (book_valid) {
              upd_mgr_.PublishBook();
            } else {
              LOG(ERROR) << "[Bitbank Feed Processor] wrong snapshot: " << packet.topic_name << ": "
                         << packet.payload.data();
            }
          }
        }
      } else {
        LOG(INFO) << "[Bitbank Feed Processor] unknown msg for " << packet.topic_name << ": "
                  << packet.payload.data();
      }
    }
  } else if (document.HasMember("url")) {
    std::string url = document["url"].GetString();
    if (url.find("/candlestick/")) {
      // kline rest, https://public.bitbank.cc/btc_jpy/candlestick/1day/2021?page=1
      std::string base_url = url.substr(0, url.find("?"));
      auto splited_url = Split(base_url, "/");
      size_t len = splited_url.size();
      CHECK_GE(len, 5);
      const std::string& interval_str = splited_url[len - 2];
      const std::string& native_symbol = splited_url[len - 4];
      if (!CheckChannelFilter("/candlestick/" + interval_str)) {
        return;
      }

      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
          return;
      }

      auto interval = BitbankFeedUpdateParser::ParseKlineInterval(interval_str);
      auto* kline = upd_mgr_.GetKlineMutable(native_symbol, interval);
      if (kline != nullptr) {
        bool to_publish = false;
        // query is split up several requests
        int page = -1;
        auto parameters = base::feed::FeedParsingProcessor::GetRequestParameters(url);
        if (parameters.count("page") > 0) {
          page = std::stod(parameters["page"]);
        }
        if (page == 0) {
          // first long query
          kline->SnapshotClear();
        } else {
          kline->UpdateClear();
        }

        int64_t timestamp = document["data"]["data"]["timestamp"].GetInt64() * 1'000'000LL;
        const auto& items = document["data"]["data"]["candlestick"].GetArray()[0]["ohlcv"];
        for (const auto& kline_data : items.GetArray()) {
          bool freezed = BitbankFeedUpdateParser::Parse1Kline(
              kline_data, kline, timestamp, interval);
          to_publish = to_publish || freezed;
        }

        if (to_publish) {
          upd_mgr_.PublishKline(native_symbol);
        }
      }
    }
  }

  upd_mgr_.onEndPacket(packet);
}  // namespace coin2::exchange::bitbank::feed_v1::impl
}  // namespace coin2::exchange::bitbank::feed_v1::impl
